package task

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"go-task/common"
	"go-task/data"
	"go-task/domain/model"
	"go.uber.org/zap"
	"net"
	"sync"
	"time"
)

var (
	TaskManager *TaskServer
)

type TaskServer struct {
	port     int
	log      *zap.SugaredLogger
	taskMap  map[string]*TaskClient
	connMap  map[net.Conn]string
	lock     sync.RWMutex
	authMap  map[net.Conn]bool
	authLock sync.RWMutex
}

type ClientData struct {
	Name string
	Conn net.Conn
}

type TaskClient struct {
	CurrentIndex int
	Clients      []*ClientData
}

func NewTaskServer(port int, log *zap.SugaredLogger) *TaskServer {
	return &TaskServer{
		port:    port,
		log:     log,
		taskMap: make(map[string]*TaskClient),
		connMap: make(map[net.Conn]string),
		authMap: make(map[net.Conn]bool),
	}
}

func (s *TaskServer) Start() {
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", s.port))
	if err != nil {
		s.log.Errorf("task task error: %v", err)
	}
	defer listener.Close()
	s.log.Infof("task task listening on port: %d", s.port)

	// 接受连接循环
	for {
		conn, err := listener.Accept()
		if err != nil {
			// 检查是否是因关闭监听器导致的错误
			if opErr, ok := err.(*net.OpError); ok && opErr.Op == "accept" {
				break // 优雅退出
			}
			s.log.Errorf("task conn accept error: %v", err)
			continue
		}
		s.log.Infof("new task client conn. host:%s", conn.RemoteAddr())
		go s.handleConnection(conn) // 启动goroutine处理连接
	}
}

func (s *TaskServer) removeConn(conn net.Conn) {
	s.lock.Lock()
	defer s.lock.Unlock()
	taskGroup, ok := s.connMap[conn]
	if !ok {
		return
	}
	delete(s.connMap, conn)
	taskClient := s.taskMap[taskGroup]
	if taskClient == nil || len(taskClient.Clients) == 0 {
		return
	}

	for index, clientData := range taskClient.Clients {
		if clientData.Conn == conn {
			taskClient.Clients = append(taskClient.Clients[:index], taskClient.Clients[index+1:]...)
		}
	}
}

func (s *TaskServer) handleConnection(conn net.Conn) {
	defer func(conn net.Conn) {
		s.log.Infof("close task client conn. host:%s", conn.RemoteAddr())
		s.removeConn(conn)
		err := conn.Close()
		if err != nil {
			s.log.Errorf("task conn close error: %v", err)
		}
	}(conn) // 确保连接关闭

	// 设置认证超时（10秒未完成认证则断开）
	err := conn.SetReadDeadline(time.Now().Add(10 * time.Second))
	if err != nil {
		s.log.Errorf("task conn set read deadline error: %v", err)
		return
	}

	var buffer bytes.Buffer
	readBuf := make([]byte, 4096)
	for {
		n, err := conn.Read(readBuf)
		if err != nil {
			s.log.Errorf("Read error:%v", err)
			return
		}

		// 将读取的数据追加到缓冲区
		buffer.Write(readBuf[:n])

		// 处理缓冲区中的数据
		for {
			// 检查是否足够读取长度头
			if buffer.Len() < 4 {
				break
			}

			// 读取长度头（大端序）
			lengthBytes := buffer.Bytes()[:4]
			length := binary.BigEndian.Uint32(lengthBytes)

			// 检查是否已接收完整数据包
			totalLength := int(length) + 4
			if buffer.Len() < totalLength {
				break
			}

			// 提取完整数据包
			packet := buffer.Next(totalLength)
			msgData := string(packet[4:])

			// 处理业务数据
			s.log.Infof("Received: %s", msgData)
			s.handleMsg(conn, msgData)
		}
	}
}

// 发送带长度头的消息
func (s *TaskServer) sendData(conn net.Conn, data []byte) {
	length := uint32(len(data))
	lengthBytes := make([]byte, 4)
	binary.BigEndian.PutUint32(lengthBytes, length)

	// 拼接长度头和实际数据
	packet := append(lengthBytes, data...)

	if _, err := conn.Write(packet); err != nil {
		s.log.Errorf("Write error:%v", err)
	}
}

func (s *TaskServer) handleMsg(conn net.Conn, msg string) {
	var msgData data.MsgData
	common.ParseJson(msg, &msgData)
	if !s.authCheck(conn) && data.CONN != msgData.Action {
		s.authFail(conn)
		s.log.Infof("task no auth will close conn. host:%s", conn.RemoteAddr())
		err := conn.Close()
		if err != nil {
			s.log.Errorf("task conn close error: %v", err)
			return
		}
		return
	}
	taskHandler, ok := taskHandlerFactory[msgData.Action]
	if !ok {
		return
	}
	taskHandler.Handler(s, conn, &msgData)
}

func (s *TaskServer) sendTimeTask(timeTask *model.TimeTask) {
	subType := timeTask.SubType
	taskData := &data.TaskData{
		TaskId:    timeTask.TaskId,
		TaskGroup: timeTask.TaskGroup,
		TaskName:  timeTask.TaskName,
		Data:      timeTask.TaskData,
	}

	if len(timeTask.TaskExtendParam) > 0 {
		var extendParam map[string]string
		common.ParseJson(timeTask.TaskExtendParam, &extendParam)
		taskData.ExtendParam = extendParam
	}

	timeTaskHis := &model.TimeTaskHis{
		TaskId:          timeTask.TaskId,
		TaskGroup:       timeTask.TaskGroup,
		TaskName:        timeTask.TaskName,
		TaskType:        timeTask.TaskType,
		CronExpr:        timeTask.CronExpr,
		SubType:         timeTask.SubType,
		TaskData:        timeTask.TaskData,
		TaskExtendParam: timeTask.TaskExtendParam,
		ExecStatus:      0,
		SuccessNum:      0,
		FailNum:         0,
		CreateTime:      common.NowStr(),
		UpdateTime:      common.NowStr(),
	}
	var timeTaskHisDao model.TimeTaskHis
	err := timeTaskHisDao.AddTaskHis(timeTaskHis)
	if err != nil {
		s.log.Errorf("task time task his error: %v", err)
	}
	taskData.TaskHisId = timeTaskHis.Id

	msgData := &data.MsgData{
		Action: data.TASK,
		Data:   common.JsonString(taskData),
	}

	s.lock.Lock()
	defer s.lock.Unlock()
	taskClient, ok := s.taskMap[timeTask.TaskGroup]
	if !ok {
		return
	}

	if common.SHARED == subType {
		for index, clientData := range taskClient.Clients {
			s.sendData(clientData.Conn, common.JsonBytes(msgData))
			if index+1 == len(taskClient.Clients) {
				taskClient.CurrentIndex = index
			}
		}
		return
	}

	if common.EXCLUSIVE == subType {
		if len(taskClient.Clients) == 1 {
			s.sendData(taskClient.Clients[0].Conn, common.JsonBytes(msgData))
			return
		}

		targetIndex := taskClient.CurrentIndex % len(taskClient.Clients)
		clientData := taskClient.Clients[targetIndex]
		s.sendData(clientData.Conn, common.JsonBytes(msgData))

		// 更新索引
		taskClient.CurrentIndex = (targetIndex + 1) % len(taskClient.Clients)
		return
	}
}

func (s *TaskServer) SendTask(taskGroup string, taskType string, data *data.MsgData) {
	s.lock.Lock()
	defer s.lock.Unlock()
	taskClient, ok := s.taskMap[taskGroup]
	if !ok {
		return
	}
	if common.SHARED == taskType {
		for index, clientData := range taskClient.Clients {
			s.sendData(clientData.Conn, common.JsonBytes(data))
			if index+1 == len(taskClient.Clients) {
				taskClient.CurrentIndex = index
			}
		}
		return
	}

	if common.EXCLUSIVE == taskType {
		if len(taskClient.Clients) == 1 {
			s.sendData(taskClient.Clients[0].Conn, common.JsonBytes(data))
			return
		}

		targetIndex := taskClient.CurrentIndex % len(taskClient.Clients)
		clientData := taskClient.Clients[targetIndex]
		s.sendData(clientData.Conn, common.JsonBytes(data))

		// 更新索引
		taskClient.CurrentIndex = (targetIndex + 1) % len(taskClient.Clients)
		return
	}
}

func (s *TaskServer) authCheck(conn net.Conn) bool {
	s.authLock.RLock()
	defer s.authLock.RUnlock()
	_, authOk := s.authMap[conn]
	return authOk
}

func (s *TaskServer) authFail(conn net.Conn) {
	resp := data.MsgData{
		Action: data.AUTH_FAIL,
		Data:   "auth failed",
	}
	s.sendData(conn, common.JsonBytes(resp))
}
